#ifndef CHANNEL_H
#define CHANNEL_H
#include "../common/message.pb.h"
#include "../common/Log.hpp"
#include "../common/Helper.hpp"
#include "../common/project.pb.h"
#include "Consumer.hpp"
#include <muduo/net/TcpConnection.h>
#include <muduo/proto/codec.h>
#include <muduo/proto/dispatcher.h>
#include <mutex>
#include <condition_variable>
#include <unordered_map>
using namespace std;

// 消息的指针
typedef shared_ptr<google::protobuf::Message> MessagePtr;
// 协议解析器
using ProtobufCodecPtr = shared_ptr<ProtobufCodec>;
// 消息的推送
using basicConsumeResponsePtr = shared_ptr<MessageQueue::basicConsumeResponse>;
// 通用响应
using basicCommonResponsePtr = shared_ptr<MessageQueue::basicCommonResponse>;

// 信道描述对象
class Channel 
{
public:
    using ptr = shared_ptr<Channel>;
    Channel(const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec)
        : _cid(UUIDHelper::uuid()), _conn(conn), _codec(codec) 
    {}

    ~Channel() 
    { 
        basicCancel(); 
    }

    string cid()
    { 
        return _cid; 
    }
    // 打开信道
    bool openChannel()
    {
        string rid = UUIDHelper::uuid();
        MessageQueue::openChannelRequest req; 
        req.set_rid(rid);
        req.set_cid(_cid);
        // 把请求信息发送出去，等待响应
        _codec->send(_conn, req);
        basicCommonResponsePtr resp = waitResponse(rid);
        return resp->ok();
    }
    // 关闭信道
    void closeChannel()
    {
        string rid = UUIDHelper::uuid();
        MessageQueue::closeChannelRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        _codec->send(_conn, req);
        waitResponse(rid);
    }
    // 声明交换机
    bool declareExchange(const string &name,
        MessageQueue::ExchangeType type, 
        bool durable, 
        bool auto_delete,
        google::protobuf::Map<string, string> &args) 
    {
        //构造一个声明虚拟机的请求对象，
        string rid = UUIDHelper::uuid();
        MessageQueue::declareExchangeRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_exchange_name(name);
        req.set_exchange_type(type);
        req.set_durable(durable);
        req.set_auto_delete(auto_delete);
        req.mutable_args()->swap(args);
        // 然后向服务器发送请求
        _codec->send(_conn, req);
        // 等待服务器的响应
        basicCommonResponsePtr resp = waitResponse(rid);
        // 返回
        return resp->ok();
    }
    // 删除交换机
    void deleteExchange(const string &name) 
    {
        string rid = UUIDHelper::uuid();
        MessageQueue::deleteExchangeRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_exchange_name(name);
        _codec->send(_conn, req);
        waitResponse(rid);
    }
    // 声明队列
    bool declareQueue(const string &qname, 
        bool qdurable, 
        bool qexclusive,
        bool qauto_delete,
        google::protobuf::Map<string, string> &qargs) 
    {
        string rid = UUIDHelper::uuid();
        MessageQueue::declareQueueRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_queue_name(qname);
        req.set_durable(qdurable);
        req.set_auto_delete(qauto_delete);
        req.set_exclusive(qexclusive);
        req.mutable_args()->swap(qargs);
        _codec->send(_conn, req);
        basicCommonResponsePtr resp = waitResponse(rid);
        return resp->ok();
    }
    // 删除队列
    void deleteQueue(const string &qname) 
    {
        string rid = UUIDHelper::uuid();
        MessageQueue::deleteQueueRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_queue_name(qname);
        _codec->send(_conn, req);
        waitResponse(rid);
    }
    // 队列的绑定
    bool queueBind(const string &ename, 
        const string &qname, 
        const string &key) 
    {
        string rid = UUIDHelper::uuid();
        MessageQueue::queueBindRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_exchange_name(ename);
        req.set_queue_name(qname);
        req.set_binding_key(key);
        _codec->send(_conn, req);
        basicCommonResponsePtr resp = waitResponse(rid);
        return resp->ok();
    }
    // 解除队列绑定
    void queueUnBind(const string &ename, const string &qname) 
    {
        string rid = UUIDHelper::uuid();
        MessageQueue::queueUnBindRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_exchange_name(ename);
        req.set_queue_name(qname);
        _codec->send(_conn, req);
        waitResponse(rid);
    }
    // 消息发布
    void basicPublish(const string &ename,
        const MessageQueue::BasicProperties *bp,
        const string &body)
    {
        string rid = UUIDHelper::uuid();
        MessageQueue::basicPublishRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_body(body);
        req.set_exchange_name(ename);
        // 设置响应的属性
        if (bp != nullptr) 
        {
            req.mutable_properties()->set_id(bp->id());
            req.mutable_properties()->set_delivery_mode(bp->delivery_mode());
            req.mutable_properties()->set_routing_key(bp->routing_key());
        }
        _codec->send(_conn, req);
        waitResponse(rid);
    }
    // 等待消息确认
    void basicAck(const string &msgid) 
    {
        if (_consumer.get() == nullptr) 
        {
            lg(Debug, "消息确认时，找不到消费者信息！");
            return ;
        }
        string rid = UUIDHelper::uuid();
        MessageQueue::basicAckRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_queue_name(_consumer->qname);
        req.set_message_id(msgid);
        _codec->send(_conn, req);
        waitResponse(rid);
    }
    // 订阅的取消
    void basicCancel() 
    {
        if (_consumer.get() == nullptr) 
            return;
        string rid = UUIDHelper::uuid();
        MessageQueue::basicCancelRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_queue_name(_consumer->qname);
        req.set_consumer_tag(_consumer->tag);
        _codec->send(_conn, req);
        waitResponse(rid);
        _consumer.reset();
    }
    // 队列的订阅
    bool basicConsume(const string &consumer_tag,
        const string &queue_name,
        bool auto_ack,
        const ConsumerCallback &cb) 
    {
        if (_consumer.get() != nullptr)
        {
            lg(Debug, "当前信道已订阅其他队列消息！");
            return false;
        }
        string rid = UUIDHelper::uuid();
        MessageQueue::basicConsumeRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_queue_name(queue_name);
        req.set_consumer_tag(consumer_tag);
        req.set_auto_ack(auto_ack);
        _codec->send(_conn, req);
        basicCommonResponsePtr resp =  waitResponse(rid);
        if (resp->ok() == false) 
        {
            lg(Debug, "添加订阅失败！");
            return false;
        }
        _consumer = make_shared<Consumer>(consumer_tag, queue_name, auto_ack, cb);
        return true;
    }

public:
    // 连接收到基础响应后，向hash_map中添加响应
    void putBasicResponse(const basicCommonResponsePtr& resp) 
    {
        unique_lock<mutex> lock(_mutex);
        _basic_resp.insert(make_pair(resp->rid(), resp));
        _cv.notify_all();
    }
    
    // 连接收到消息推送后，需要通过信道找到对应的消费者对象，通过回调函数进行消息处理
    void consume(const basicConsumeResponsePtr& resp)
    {
        if (_consumer.get() == nullptr) 
        {
            lg(Debug, "消息处理时，未找到订阅者信息！");
            return;
        }
        if (_consumer->tag != resp->consumer_tag()) 
        {
            lg(Debug, "收到的推送消息中的消费者标识，与当前信道消费者标识不一致！");
            return ;
        }
        _consumer->callback(resp->consumer_tag(), resp->mutable_properties(), resp->body());
    }

private:
    // 这是一个通用的等待响应，所有的响应都在这里等待，根据ResponseID来进行等待指定的Response
    basicCommonResponsePtr waitResponse(const string &rid) 
    {
        unique_lock<mutex> lock(_mutex);
        // 这里要等待指定请求的响应来实现同步
        _cv.wait(lock, [&rid, this](){
            return _basic_resp.find(rid) != _basic_resp.end();
        });
        // while(condition()) _cv.wait();
        basicCommonResponsePtr basic_resp = _basic_resp[rid];
        _basic_resp.erase(rid);
        // 如果等到了，就把通用响应发送回去
        return basic_resp;
    }

private:
    string _cid;
    muduo::net::TcpConnectionPtr _conn;
    ProtobufCodecPtr _codec;
    Consumer::ptr _consumer;
    mutex _mutex;
    condition_variable _cv;
    // 请求对应的响应消息队列
    unordered_map<string, basicCommonResponsePtr> _basic_resp;
};

// 信道的管理
class ChannelManager 
{
public:
    using ptr = shared_ptr<ChannelManager>;
    ChannelManager()
    {}
    // 创建信道
    Channel::ptr create(const muduo::net::TcpConnectionPtr &conn, 
        const ProtobufCodecPtr &codec) 
    {
        unique_lock<mutex> lock(_mutex);
        auto channel = make_shared<Channel>(conn, codec);
        _channels.insert(make_pair(channel->cid(), channel));
        return channel;
    }
    // 移除信道
    void remove(const string &cid) 
    {
        unique_lock<mutex> lock(_mutex);
        _channels.erase(cid);
    }
    // 查询信道
    Channel::ptr get(const string &cid) 
    {
        unique_lock<mutex> lock(_mutex);
        auto it = _channels.find(cid);
        if (it == _channels.end()) 
            return Channel::ptr();
        return it->second;
    }

private:
    mutex _mutex;
    unordered_map<string, Channel::ptr> _channels;
};

#endif